#!/usr/bin/env python
# -*- coding: UTF-8 -*-
'''
@Project ：sumo_jingtai
@File ：离线数据发送.py
@IDE ：PyCharm
@Author ：xinyingjie
@Date ：2025/6/13 8:41
'''
import json
import time
from datetime import datetime

from kafka import KafkaProducer
from loguru import logger

from utils.config_temp import KAFKA_HOST, RADAR_DATA_LIST_CJ

station_orgCode_dict = {11: "e1_data_CJ-10-01"}

kafka_producer = KafkaProducer(bootstrap_servers=KAFKA_HOST, key_serializer=str.encode,
                               value_serializer=lambda x: json.dumps(x).encode('utf-8'))
topic_list = ['e1_data_CJ-10-01-0', 'e1_data_CJ-10-02-0', 'e1_data_JJ-10-01-0', 'e1_data_JJ-10-02-0']


def get_timestamp_milliseconds(t_str):
    """
    将ISO 8601时间戳字符串转为13位毫秒级时间戳
    :param t_str: ISO 8601时间戳字符串
    :return: 13位毫秒级时间戳
    """
    # t_str = data['timestamp']
    # 解析 ISO 8601 时间字符串为 datetime 对象
    dt = datetime.strptime(t_str, "%Y-%m-%d %H:%M:%S:%f")
    # 将 datetime 对象转换为秒级时间戳
    timestamp_seconds = dt.timestamp()
    # 将秒级时间戳转换为毫秒级时间戳
    timestamp_milliseconds = int(timestamp_seconds * 1000)
    return timestamp_milliseconds


def send_by_sn():
    file_path = r"G:\Log\merged_CJ20050624_intersection.jsonl"
    send_radar_topic = "CJ-10-SUM"
    send_kako_topic = "send_kako_topic_CJ"
    before_time = None
    with open(file_path, 'r', encoding='utf-8') as f:
        while True:
            line = f.readline().strip()
            if line:
                data = json.loads(line)
                # logger.info(f"{send_topic} {data}")
                if "globalTime" in data.keys():
                    kafka_producer.send(send_radar_topic, data, key=str(time.time()))
                    time_now = get_timestamp_milliseconds(data['globalTime'])
                else:
                    kafka_producer.send(send_kako_topic, data, key=str(time.time()))
                    time_now = data['timestamp']
                # kafka_producer.send(send_topic, data, key=str(time.time()))
                # globalTime = data['globalTime']
                if before_time:
                    if before_time - time_now > 0:
                        logger.error("时间错误")
                    sleep_time = abs(before_time - time_now) / 1000
                    time.sleep(sleep_time)
                before_time = time_now
            else:
                break


send_by_sn()
